package com.ilikesoup.instant.executor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.*;

/**
 * 该线程池实现支持，对于包含相同互斥键任务的有序执行。
 * 支持 {@link KeyPickRunnable} , {@link KeyPickCallable} 和 {@link KeyPickRunnableFuture}，
 * {@link KeyPickRunnable} 和 {@link KeyPickCallable} 最终会在 {@link #newTaskFor} 方法中转换为 {@link KeyPickRunnableFuture}
 * 通过 {@link KeyPickRunnableFuture#getKeys} 获取的互斥键集合来实现有序执行。
 * <p>
 * 在将任务提交给线程池前，判断如果为 {@link KeyPickRunnableFuture} 并且它携带了互斥键，就加入等待队列，并唤醒扫描线程。
 * 扫描线程会遍历等待队列，判断它的互斥键是否已经完成执行，如果完成了，那么将任务提交给线程池处理。但它不能比排在它之前的互斥任务先执行。
 * <p>
 * 在有任务加入等待队列或有任务已经执行完成时，会唤醒扫描线程；扫描线程等待超时会自己苏醒 {@link Object#wait(long)}，
 * 可以通过 scannerThreadAwakeTimeout构造参数设置这个自动苏醒时间间隔。
 * <p>
 * 只有一个线程会向运行键集中新增互斥键，但是可能会有多个(池中的)线程从其中移除。
 * <p>
 * 该线程池扩展能够避免池中线程的在互斥操作时的等待过程，将多线程间的互斥判断提前到处理之前，如果不满足互斥条件就不交给线程池处理，可以让其他满足条件的先执行。
 * 它适用于：1 在执行任务前就已经知道了任务关联的互斥键； 2 同一互斥键，必须按提交顺序处理。
 *
 * @author soup
 * @see KeyPickCallable
 * @see KeyPickRunnable
 * @see KeyPickRunnableFuture
 */
@SuppressWarnings({"rawtypes", "unchecked"})
public abstract class AbstractKeyPickThreadPoolExecutor extends ThreadPoolExecutor {

    private static final Logger logger = LoggerFactory.getLogger(AbstractKeyPickThreadPoolExecutor.class);

    /**
     * 等待队列，任务的互斥键集中存在正在运行的键，在此队列中进行等待
     */
    private final BlockingQueue<KeyPickRunnableFuture> waitingQueue;
    /**
     * 等待队列扫描线程
     * 扫描等待队列，通过任务携带的key确定哪些任务可以交给线程池执行，哪些需要继续等待。
     * 扫描完成一轮后会继续等待直至被唤醒或等待超时
     */
    private final ScannerThread waitingQueueScanner;

    @Override
    public void execute(Runnable command) {
        if (command instanceof KeyPickRunnableFuture) {
            KeyPickRunnableFuture task = (KeyPickRunnableFuture) command;
            Collection keys = task.getKeys();
            if (keys != null && keys.size() > 0) {
                if(waitingQueue.offer(task)) {
                    waitingQueueScanner.signal();
                    return;
                }
                getRejectedExecutionHandler().rejectedExecution(command, this);
            }
        }
        super.execute(command);
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        if(callable instanceof KeyPickCallable) {
            return new KeyPickRunnableFuture(callable, ((KeyPickCallable) callable).getKeys());
        }
        return super.newTaskFor(callable);
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        if (runnable instanceof KeyPickRunnableFuture && value == null) {
            return (KeyPickRunnableFuture) runnable;
        }
        if (runnable instanceof KeyPickRunnable) {
            return new KeyPickRunnableFuture<>((KeyPickRunnable) runnable, value);
        }
        return super.newTaskFor(runnable, value);
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        if (r instanceof KeyPickRunnableFuture) {
            KeyPickRunnableFuture task = (KeyPickRunnableFuture) r;
            removeKeys(task.getKeys());
            logger.debug("线程池队列任务数：{}，等待队列任务数：{}", getQueue().size(), waitingQueue.size());
        }
        waitingQueueScanner.signal();
        if (t != null) {
            logger.error("执行任务失败", t);
        }
    }

    private class ScannerThread extends Thread {
        /**
         * 用于线程的等待和唤醒
         */
        final Object monitor = new Object();
        /**
         * 唤醒超时时间 单位ms，0-不自动唤醒
         */
        final long awakeTimeout;
        /**
         * 是否正在运行
         */
        volatile boolean running = true;

        ScannerThread(){
            this("WaitingQueueScan-Task");
        }

        ScannerThread(String name) {
            this(name, 30_000);
        }

        ScannerThread(String name, long awakeTimeout) {
            super(name);
            setDaemon(true);
            this.awakeTimeout = awakeTimeout < 0 ? 0 : awakeTimeout;
        }

        @Override
        public void run() {
            synchronized (monitor) {
                try {
                    monitor.wait();
                } catch (InterruptedException e) {
                    // ignore
                }
            }
            while (running) {
                if (waitingQueue.size() > 0) {
                    KeyPickRunnableFuture task = null;
                    try {
                        Set<Object> waitingKeySet = new HashSet<>();
                        Iterator<KeyPickRunnableFuture> iterator = waitingQueue.iterator();
                        while (iterator.hasNext()) {
                            task = iterator.next();
                            Collection keys = task.getKeys();
                            if (keys.stream().noneMatch(waitingKeySet::contains) && addKeysIfNotExists(keys)) {
                                try {
                                    AbstractKeyPickThreadPoolExecutor.super.execute(task);
                                } catch (RejectedExecutionException e) {
                                    removeKeys(keys);
                                    throw e;
                                }
                                iterator.remove();
                            } else {
                                waitingKeySet.addAll(keys);
                            }
                        }
                    } catch (RejectedExecutionException e) {
                        // ignore 执行线程池队列满了，过一会再消费
                    } catch (Exception e) {
                        logger.error("执行任务时：{}，发生异常。", task, e);
                    }
                }
                synchronized (monitor) {
                    try {
                        do {
                            monitor.wait(awakeTimeout);
                        } while (getQueue().remainingCapacity() == 0 || waitingQueue.isEmpty());
                    } catch (InterruptedException e) {
                        // ignore
                    }
                }
            }
            logger.debug("扫描等待队列线程关闭");
        }

        void terminate() {
            running = false;
            signal();
        }

        void signal() {
            synchronized (monitor) {
                monitor.notify();
            }
        }
    }

    /**
     * 从运行中的键集中移除键集
     * @param keys
     */
    protected abstract void removeKeys(Collection<?> keys);

    /**
     * 如果运行中的键集不存在，就向其中添加键集
     * @param keys
     * @return
     */
    protected abstract boolean addKeysIfNotExists(Collection<?> keys);

    /**
     * 清除已存在的键
     */
    public abstract void clearKeys();

    @Override
    protected void terminated() {
        super.terminated();
        waitingQueueScanner.terminate();
    }

    /**
     * 需要子类在创建完成对象后，调用该方法，启动等待队列扫描线程
     */
    protected void start() {
        waitingQueueScanner.start();
    }

    protected AbstractKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        waitingQueue = new LinkedBlockingQueue<>();
        waitingQueueScanner = new ScannerThread();
    }

    protected AbstractKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        waitingQueue = new LinkedBlockingQueue<>();
        waitingQueueScanner = new ScannerThread();
    }

    protected AbstractKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        waitingQueue = new LinkedBlockingQueue<>();
        waitingQueueScanner = new ScannerThread();
    }

    protected AbstractKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        waitingQueue = new LinkedBlockingQueue<>();
        waitingQueueScanner = new ScannerThread();
    }

    protected AbstractKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                                RejectedExecutionHandler handler, String scannerThreadName, long scannerThreadAwakeTimeout) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        waitingQueue = new LinkedBlockingDeque<>();
        waitingQueueScanner = new ScannerThread(scannerThreadName, scannerThreadAwakeTimeout);
    }

    protected AbstractKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                                RejectedExecutionHandler handler, BlockingQueue<KeyPickRunnableFuture> waitingQueue, String scannerThreadName, long scannerThreadAwakeTimeout) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        this.waitingQueue = waitingQueue;
        waitingQueueScanner = new ScannerThread(scannerThreadName, scannerThreadAwakeTimeout);
    }

    public static final RejectedExecutionHandler THROW_REJECTED_EXECUTION_HANDLER = new ThrowRejectedExecutionHandler();

    protected AbstractKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, BlockingQueue<Runnable> workQueue, BlockingQueue<KeyPickRunnableFuture> waitingQueue, String scannerThreadName) {
        super(corePoolSize, maximumPoolSize, 1, TimeUnit.MINUTES, workQueue, new CustomizableThreadFactory(scannerThreadName+"-"), THROW_REJECTED_EXECUTION_HANDLER);
        this.waitingQueue = waitingQueue;
        waitingQueueScanner = new ScannerThread(scannerThreadName);
    }

}
